import { bootEvent } from '../boot';
import os from 'os';
import fs from 'fs';
import path from 'path';
import * as http from 'http';
import { Config, loadConfig } from '../app';
import { makeLabelLogger } from '../util/logger';
import * as sio from 'socket.io';
import * as express from 'express';
import { createHandyClient, IHandyRedis } from 'handy-redis';
import redis from 'redis';
import { sleep, getRandomArbitrary } from '../util/tools';
import { EventEmitter } from 'events';
import { Container } from 'inversify';
import { IUserService, tsIUserService } from 'src/service/api/user';
import { UserServiceImpl } from 'src/service/impl/userService';

let { llog, lerror } = makeLabelLogger('mbsync');
export let syncRedisWait: IHandyRedis;
export let syncRedisCommon: IHandyRedis;
export let msgEvent = new EventEmitter();
export let DataCenterUrl = 'http://have.you.define.ask:9527';

require('./event-user');
require('./usmgr');
require('../util/pull');

export let container: Container;
export let startSyncServer = async function() {
	await loadConfig();

	DataCenterUrl = process.env.DataCenterUrl || DataCenterUrl;

	let port = Config.getSyncPort();
	llog(`* sync server is listening on port : ${port}`);
	llog('DataCenterUrl', DataCenterUrl);

	let app = require('express')() as express.Application;
	let httpServer = http.createServer(app);

	container = new Container();
	container
		.bind<IUserService>(tsIUserService)
		.to(UserServiceImpl)
		.inSingletonScope();

	let us = container.get<IUserService>(tsIUserService);
	llog('us');

	httpServer.listen(port);
	WatchRecv();

	bootEvent.emit('appInit', app);
};

let WatchRecv = async () => {
	let config = Config.getSyncRedis();
	syncRedisCommon = createHandyClient(config);
	syncRedisCommon.redis.on('error', () => {
		if ((<any>syncRedisCommon).errorCount++ == 0) {
			llog('redis/common disconnect');
		}
	});
	syncRedisCommon.redis.on('end', () => {
		llog('redis/common end');
	});
	syncRedisCommon.redis.on('connect', () => {
		llog('redis/common connect');
		(<any>syncRedisCommon).errorCount = 0;
	});

	config = Object.assign(config, { enable_offline_queue: false });
	syncRedisWait = createHandyClient(config);
	syncRedisWait.redis.on('error', () => {
		if ((<any>syncRedisWait).errorCount++ == 0) {
			llog('redis/wait disconnect');
		}
	});
	let fillConnect;
	syncRedisWait.redis.on('connect', () => {
		llog('redis/wait connect', config);
		(<any>syncRedisWait).errorCount = 0;
	});
	syncRedisWait.redis.on('ready', () => {
		llog('redis/wait ready');
		fillConnect();
	});
	await new Promise(fill => (fillConnect = fill));
	await syncRedisWait.set('start_time', new Date().toLocaleString());
	let k = 0;

	syncRedisCommon.set('processed:fromLaunch', (0).toString());
	while (true) {
		let msg;
		try {
			// llog('wait msg', k++);
			[, msg] = await syncRedisWait.blpop(['mq-sync'], 0);
		} catch (e) {
			if (!syncRedisWait.redis.connected) {
				llog('get msg fail', e.message);
				await new Promise(fill => (fillConnect = fill));
				continue;
			}
		}

		try {
			let jsonmsg = JSON.parse(msg);
			llog('emit msg', jsonmsg.type, jsonmsg.srcHost);
			msgEvent.emit(jsonmsg.type, jsonmsg);
		} catch (e) {
			lerror('process msg fail', e);
		}

		syncRedisWait.incr('processed:total');
		syncRedisWait.incr('processed:fromLaunch');
	}
};

export let redisXAdd = async (redis, ...args) => {
	let retryCount = 3;
	while (true) {
		try {
			return await new Promise((fill, reject) => {
				redis.xadd(...args, function(err, reply) {
					if (err) reject(err);
					else fill(reply);
				});
			});
		} catch (e) {
			lerror('redisXAdd fail', e, 'retryCount', retryCount);
			if (--retryCount < 0) throw e;
		}
	}
};

export let publishMessage = async (type: string, action: any) => {
	return await redisXAdd(syncRedisCommon.redis, 'ub', '*', 'type', type, 'msg', JSON.stringify(action));
};
